package com.ruyuan.event.streaming.process;
import com.ruyuan.event.streaming.pojo.EventServerLog;
import com.ruyuan.event.streaming.utils.Constants;
import com.ruyuan.event.streaming.utils.HBaseUtils;
import com.ruyuan.event.streaming.utils.RedisUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.HTable;
import redis.clients.jedis.Jedis;

public class EventServerLogRichFlatMap extends RichFlatMapFunction<EventServerLog,String> {
    String tableName = Constants.HBASE_EVENT_SERVER_LOG_TABLE;
    HTable hTable;
    Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hTable = HBaseUtils.initHbaseClient(tableName);
        jedis = RedisUtils.getJedis();
        super.open(parameters);
    }

    /**
     * 写入serverLog的数据到redis缓存和HBase
     * redis的key -> request_id+event_id+event_user_id.
     * */
    @Override
    public void flatMap(EventServerLog eventServerLog, Collector<String> out) {
        byte[] key = ProcessETLUtils.generateBytesKey(eventServerLog);
        EventServerLog eventSerLog = ProcessETLUtils.generateServerContext(eventServerLog);
        ProcessETLUtils.writeRedis(jedis,key,eventSerLog);
        ProcessETLUtils.writeHbase(hTable,key,eventSerLog);
    }
}